package query;

import index.HashIndex;

import java.util.ArrayList;
import java.util.HashMap;

import global.AttrOperator;
import global.AttrType;
import global.Minibase;
import global.SearchKey;
import heap.HeapFile;
import parser.AST_Select;
import relop.*;

/**
 * Execution plan for selecting tuples.
 */
/*
 * For Select, you will implement a basic query optimizer. The parser will give
 * you an array of table names to select from, an array of (unique) column names
 * to project, and an array of selection predicates (in conjunctive normal
 * form). The default plan is to use FileScans and SimpleJoins for all the
 * tables, add a Selection for each conjunct, and have one Projection at the
 * root of the Iterator tree. For example:
 */

// EXPLAIN SELECT d, a FROM T1, T2 WHERE a = c and b = d or a = 5;

class Select implements Plan {

	/**
	 * Optimizes the plan, given the parsed query.
	 * 
	 * @throws QueryException
	 *             if validation fails
	 */
	protected String[] table_names;
	protected String[] column_names;
	protected Predicate[][] predicates;
	protected Projection p;
	protected Selection s;
	protected Iterator final_itr;
	protected boolean isExplain;
	protected boolean isDistinct;
	protected Integer[] cols;
	protected Schema newSchema;
	protected ArrayList<relop.Iterator> table_itrs_list;
	private HashMap<String, IndexDesc> columnIndexes = new HashMap<String, IndexDesc>();
	private relop.Iterator tmp;

	/*
	 * Given the tables T1(a, b) and T2(c, d) and the following query:
	 * 
	 * EXPLAIN SELECT d, a FROM T1, T2 WHERE a = c and b = d or a = 5; The
	 * default (naive) execution plan is as follows:
	 * 
	 * Projection : {3}, {0} Selection : b = d OR a = 5 Selection : a = c OR a =
	 * 5 SimpleJoin : (cross) FileScan : T2 FileScan : T1
	 */

	public static relop.Iterator get_table_itr(String table0) {
		HeapFile file = new HeapFile(table0);
		Schema schema = Minibase.SystemCatalog.getSchema(table0);
		// System.out.println("Table 2 schema " );
		// schema.print();
		return new FileScan(schema, file);
	}

	public Select(AST_Select tree) throws QueryException {
		table_names = tree.getTables();
		column_names = tree.getColumns();
		predicates = tree.getPredicates();
		isExplain = tree.isExplain;
		isDistinct = tree.isDistinct;

		for (String tabname : table_names) {
			QueryCheck.tableExists(tabname);
		}

		table_itrs_list = new ArrayList<relop.Iterator>();
		for (int i = 0; i < table_names.length; i++) {
			table_itrs_list.add(get_table_itr(table_names[i]));
			
			// Create a hash map containing indexes and the column indexed keyed on column
			IndexDesc[] desc = Minibase.SystemCatalog.getIndexes(table_names[i]);
			for(IndexDesc id : desc){
				columnIndexes.put(id.columnName, id);
			}
		}
		
		
		// java.util.Arrays.deepToString(predicates));
		/*
		 * foreach table, check in list of predicates , if any of predicate is
		 * of type a=constant AND column 'a' is one of the columns of the
		 * current tabel u r sscanning .. if such a table exists , modify that
		 * iterator to its upgraded select iterator.. Once a predicate is
		 * applied to a table, it should be removed from the list of predicates.
		 * For doing this, we add all the predicates that have not been applied
		 * except the one that has been applied.
		 */
		int table_no = -1;

		// If a predicate has been applied, add it to this hash map. This way we
		// know which predicates
		// have already been applied so that we do not apply that again.
		HashMap<String, Integer> removedPreds = new HashMap<String, Integer>();
		ArrayList<ArrayList<Predicate>> o_predicates = new ArrayList<ArrayList<Predicate>>();
		
		for (Predicate[] arrp : predicates) {
			ArrayList<Predicate> list = new ArrayList<Predicate>();
			for (Predicate p : arrp) {
				list.add(p);
			}
			o_predicates.add(list);
		}
		
		System.out.println("Original supplied predicates = " + o_predicates.toString());
		for (Iterator i : table_itrs_list) {
			table_no++;
			for (int j = 0; j < predicates.length; j++) {
				Predicate[] arrp = predicates[j];
				ArrayList<Predicate> pList = null;
				if(j < o_predicates.size()){
					pList = o_predicates.get(j);
				}
				for (Predicate p : arrp) {
					Schema s = Minibase.SystemCatalog.getSchema(table_names[table_no]);
					
						// This predicate is not in remPred. This means it has
						// not been applied yet.
						switch (p.rtype) {
						case AttrType.FLOAT:
						case AttrType.INTEGER:
						case AttrType.STRING:
							// Apply this predicate only if it has not been applied previously							
							if (p.column_validate(s)){
								if(!removedPreds.containsKey(p.toString())){
									
									// If the column in the predicate is indexed, use key scan
									if(columnIndexes.containsKey(p.left)){
										IndexDesc id = columnIndexes.get(p.left);
										HashIndex hi = new HashIndex(id.indexName);
										table_itrs_list.get(table_no).close();
										KeyScan ks = new KeyScan(s, hi, new SearchKey(p.right), new HeapFile(id.indexName));
										table_itrs_list.set(table_no, ks);
										System.out.println("Done with " + p.toString());
									}
									else{
										table_itrs_list.get(table_no).close();
										table_itrs_list.set(table_no, new Selection(i,p));
										System.out.println("Done with " + p.toString());
									}
									// Put this to remPred so that it will not be
									// applied agn
									removedPreds.put(p.toString(), 0);
								}
							}
							if(pList != null)
								pList.remove(p);							
							
							
							break;
						case AttrType.COLNAME:
						case AttrType.FIELDNO:
							// Do nothing
							break;
						default:
							System.out.println("Doom 1");
						}					
				}
				if(pList != null){
				if(pList.isEmpty()){
					o_predicates.remove(j);
				}
				else if( j < o_predicates.size())
					o_predicates.set(j, pList);
				}
				// one thing to note is that ... if p0 = a = 5 OR b=10 ... but b
				// is not in T0, in future, T1 might satisfy b=10 and hence you
				// have
				// to remove b=10 from the new p0 which you added in
				// retained_predicates; TODO
			}
		}

		// TODO: checking the passing arguments ...

		System.out.println("Modified predicates =" + o_predicates.toString()); // java.util.Arrays.deepToString(predicates));
		// System.out.println(java.util.Arrays.deepToString(column_names));

		/* marge tables here using SimpleJoin */
		final_itr = table_itrs_list.get(0); // get_table_itr(table_names[0]);
		Schema final_schema = Minibase.SystemCatalog.getSchema(table_names[0]);

		
		if (table_names.length >= 2) {
			tmp = null;

			for (int i = 1; i < table_names.length; i++) {
				tmp = table_itrs_list.get(i); // get_table_itr (table_names[i]);

				// TODO: test this if clause.
				if (Minibase.SystemCatalog
						.getTableCardinality(table_names[i - 1]) < Minibase.SystemCatalog
						.getTableCardinality(table_names[i])) {
					
					final_itr = new SimpleJoin(final_itr, tmp);
					final_schema = Schema.join(final_schema,
							Minibase.SystemCatalog.getSchema(table_names[i]));
				} else {
					final_itr = new SimpleJoin(tmp, final_itr);
					final_schema = Schema.join(Minibase.SystemCatalog
							.getSchema(table_names[i]), final_schema);
				}
			}
		}
		
		// remove this ?
		QueryCheck.predicates(final_schema, predicates);

		// what shud be done to this snippet ?
		// Selection s = null;
		if (o_predicates.size() > 0) {
			ArrayList<Predicate> predicate0 = o_predicates.remove(0);
			Predicate[] arr = new Predicate[predicate0.size()];
			predicate0.toArray(arr);
			s = new Selection(final_itr, arr);
			
			for (ArrayList<Predicate> ap : o_predicates) {
				arr = new Predicate[ap.size()];
				ap.toArray(arr); // read its documentation..
				s = new Selection(s, arr);
			}
		} else {
			// make a new always true predicate.
			Predicate alwaysTrue = new Predicate(AttrOperator.EQ,
					AttrType.FIELDNO, 1, AttrType.FIELDNO, 1);
			// new Predicate(AttrOperator.GT, AttrType.FIELDNO, 3,
			// AttrType.FLOAT, 65F),
			s = new Selection(final_itr, alwaysTrue);
		}
		
		int column_cnt = column_names.length;
		if (column_cnt > 0) {
			cols = new Integer[column_cnt];
			int i = 0;
			for (String col : column_names) {
				cols[i++] = final_schema.fieldNumber(col);
			}
		} else {
			cols = new Integer[final_schema.getCount()];
			for (int i = 0; i < final_schema.getCount(); i++) {
				cols[i] = i;
			}
		}

		// System.out.println ("final schema = " + final_schema.());
		p = new Projection(s, cols);

		// make newSchema as this will be used in printing the row count
		newSchema = new Schema(cols.length);
		int newFldCnt = 0;
		for (Integer i : cols)
			newSchema.initField(newFldCnt++, final_schema, (i));

	} // public Select(AST_Select tree) throws QueryException

	/**
	 * Executes the plan and prints applicable output.
	 */
	public void execute() {

		if (isExplain == true)
			p.explain(1);
		else {
			Tuple old_tuple;
			int rows_count = 0;
			newSchema.print(); // header
			HashMap<String, Integer> distinct = new HashMap<String, Integer>();

			while (p.hasNext()) {
				old_tuple = p.getNext();
				if (isDistinct) {
					if (distinct.containsKey(new String(old_tuple.getData())))
						continue;
				}
				old_tuple.print();
				distinct.put(new String(old_tuple.getData()), 1);
				rows_count++;
			}
			System.out.println("\n" + rows_count + " rows affected. ");

		}
		p.close();
		s.close();
		final_itr.close();
		for (Iterator i : table_itrs_list) {
			i.close();
		}
		if(tmp != null)
			tmp.close();

	} // public void execute()

} // class Select implements Plan

/*
 * Sample output for pushing select
MSQL> explain SELECT sid, points from Students, Grades where gsid = sid and sid > 10;

Original supplied predicates = [[gsid = sid], [sid > 10]]
Done with sid > 10
Modified predicates =[[gsid = sid]]
  Projection : {3}, {2}
    Selection : gsid = sid
      SimpleJoin : (cross)
        FileScan : Grades
        Selection : sid > 10
          FileScan : Students
*/
